package com.atbeijing.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atbeijing.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

/**
 * 用户跳出明细统计
 */
public class UserJumpDetailApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

//        //TODO 设置检查点
//        //5秒下发一次检查点
//        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
//        //完成一次checkpoint超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(60000);
//        //job取消后是否保存检查点
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        //重启三次,间隔时间为3000ms
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000));
//        //状态后端保存位置
//        env.setStateBackend(new FsStateBackend("hdfs://hadoop202:8020/gmall/ck"));
//        System.setProperty("HADOOP_USER_NAME","wjg");

        //TODO 从kafka中读取数据
        //声明主题和消费者组
        String topic = "dwd_page_log";
        String groupId = "user_jump_detail_app_group";
        String sinkTopic = "dwm_user_jump_detail";

        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        DataStreamSource<String> kafkaDS = env.addSource(kafkaSource);

//        DataStream<String> kafkaDS = env
//                .fromElements(
//                        "{\"common\":{\"mid\":\"101\"},\"page\":{\"page_id\":\"home\"},\"ts\":10000} ",
//                        "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"home\"},\"ts\":12000}",
//                        "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"good_list\",\"last_page_id\":" +
//                                "\"home\"},\"ts\":15000} ",
//                        "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"good_list\",\"last_page_id\":" +
//                                "\"detail\"},\"ts\":30000} "
//                );

        //TODO 设置水位线 根据mid分组
        KeyedStream<JSONObject, String> keyedDS = kafkaDS
                //转json
                .map(JSON::parseObject)
                //设置水位线
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                        return element.getLong("ts");
                                    }
                                })
                )
                //根据mid分组
                .keyBy(r -> r.getJSONObject("common").getString("mid"));

        //TODO 设置CEP模板
        Pattern<JSONObject, JSONObject> pattern = Pattern
                .<JSONObject>begin("first")
                //当前页面是访问app时的第一个页面
                .where(
                        new SimpleCondition<JSONObject>() {
                            @Override
                            public boolean filter(JSONObject value) {
                                String latsPageId = value.getJSONObject("page").getString("last_page_id");
                                if (latsPageId == null || latsPageId.length() == 0) {
                                    return true;
                                }
                                return false;
                            }
                        }
                )
                //紧接着第二条数据
                .next("second")
                .where(
                        new SimpleCondition<JSONObject>() {
                            @Override
                            public boolean filter(JSONObject value) throws Exception {
                                String pageId = value.getJSONObject("page").getString("page_id");
                                if (pageId != null && pageId.length() > 0) {
                                    return true;
                                }
                                return false;
                            }
                        }
                )
                //两个事件再10秒内发生
                .within(Time.seconds(10));

        //TODO 将模式应用到流中
        PatternStream<JSONObject> filterDS = CEP.pattern(keyedDS, pattern);

        //TODO 从流中提取超时数据
        OutputTag<String> outputTag = new OutputTag<String>("outputTag"){};

        SingleOutputStreamOperator<String> resultDS = filterDS.flatSelect(
                outputTag,
                //处理超时的数据
                new PatternFlatTimeoutFunction<JSONObject, String>() {
                    //map中key为first和second,list中为满足first和second过滤条件的数据
                    @Override
                    public void timeout(Map<String, List<JSONObject>> map, long timeoutTimestamp, Collector<String> collector) throws Exception {
                        List<JSONObject> first = map.get("first");
                        for (JSONObject jsonObject : first) {
                            //输出到侧输出流
                            collector.collect(jsonObject.toJSONString());
                        }
                    }
                },
                //处理未超时数据
                new PatternFlatSelectFunction<JSONObject, String>() {
                    @Override
                    public void flatSelect(Map<String, List<JSONObject>> map, Collector<String> collector) throws Exception {
                        //未超时数据不需要就不处理
                    }
                }
        );

        //获取侧输出流
        DataStream<String> sideOutput = resultDS.getSideOutput(outputTag);

        //TODO 将数据写到kafka的dwm层
        sideOutput.print("用户跳出>>>");

        sideOutput.addSink(MyKafkaUtil.getKafkaSink(sinkTopic));

        env.execute();
    }
}
